package com.lagou.client;

import com.lagou.client.codec.RpcClientFrameDecoder;
import com.lagou.client.codec.RpcClientFrameEncoder;
import com.lagou.client.codec.RpcClientProtocolDecoder;
import com.lagou.client.codec.RpcClientProtocolEncoder;
import com.lagou.client.handler.dispatcher.OperationResultFuture;
import com.lagou.client.handler.dispatcher.RequestPendingCenter;
import com.lagou.client.handler.dispatcher.ResponseDispatcherHandler;
import com.lagou.server.common.OperationResult;
import com.lagou.server.common.RequestMessage;
import com.lagou.server.common.service.RpcOperation;
import com.lagou.server.common.service.RpcOperationResult;
import com.lagou.util.IdUtil;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.logging.LogLevel;
import io.netty.handler.logging.LoggingHandler;
import org.apache.curator.RetryPolicy;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.cache.PathChildrenCache;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheEvent;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheListener;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.lang.reflect.InvocationHandler;
import java.lang.reflect.Method;
import java.lang.reflect.Proxy;
import java.net.InetSocketAddress;
import java.util.List;


public class Client {
    Logger logger = LoggerFactory.getLogger(Client.class);

    NioEventLoopGroup group ;
    RequestPendingCenter requestPendingCenter = new RequestPendingCenter();
    ChannelFuture clientChannel;

    CuratorFramework zkclient;
    volatile boolean isConnected = false;

    public Client(){
        //创建会话
        RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 3);
        zkclient=  CuratorFrameworkFactory
                .builder()
                .namespace("base")
                .connectString("106.12.10.223:2181") //server地址
                .sessionTimeoutMs(5000) // 会话超时时间
                .connectionTimeoutMs(3000) // 连接超时时间
                .retryPolicy(retryPolicy) // 重试策略
                .build(); //
        zkclient.start();

        group  = new NioEventLoopGroup();
        Bootstrap bootstrap = new Bootstrap();
        bootstrap.channel(NioSocketChannel.class);

        try{
            bootstrap.group(group);
            bootstrap.handler(new ChannelInitializer<NioSocketChannel>() {
                @Override
                protected void initChannel(NioSocketChannel ch) throws Exception {
                    ChannelPipeline pipeline = ch.pipeline();

                    pipeline.addLast(new RpcClientFrameDecoder());
                    pipeline.addLast(new RpcClientFrameEncoder());
                    pipeline.addLast(new RpcClientProtocolEncoder());
                    pipeline.addLast(new RpcClientProtocolDecoder());
                    pipeline.addLast(new ResponseDispatcherHandler(requestPendingCenter));
                    pipeline.addLast(new LoggingHandler(LogLevel.INFO));
                }
            });

             List<String>  children = zkclient.getChildren().forPath("/hosts");
             for(String path : children) {
                 String host = path.replace("/hosts/","");
                 String  ip = host.split(":")[0];
                 int port = Integer.parseInt(host.split(":")[1]);
                 clientChannel = bootstrap.connect(ip, port);
                 clientChannel.sync();
                 logger.info("建立连接ip:" +ip+",port:"+port);
                 ClientSessions.getInstance().addClientSession(ip+":"+port,clientChannel.channel());
             }

             isConnected = true;

            PathChildrenCache pathChildrenCache = new PathChildrenCache(zkclient,"/hosts",true);

            pathChildrenCache.getListenable().addListener(new PathChildrenCacheListener() {
                @Override
                public void childEvent(CuratorFramework curatorFramework, PathChildrenCacheEvent event) throws Exception {
                    if(event.getType().equals(PathChildrenCacheEvent.Type.CHILD_ADDED)){
                        String host = event.getData().getPath().replace("/hosts/","");
                        System.out.println("添加子节点:" + event.getData().getPath());
                        System.out.println("子节点数据:" + new String(event.getData().getData()));

                        String  ip = host.split(":")[0];
                        int port = Integer.parseInt(host.split(":")[1]);
                        clientChannel = bootstrap.connect(ip, port);
                        clientChannel.sync();
                        logger.info("建立连接ip:" +ip+",port:"+port);
                        ClientSessions.getInstance().addClientSession(ip,clientChannel.channel());


                    }else if(event.getType().equals(PathChildrenCacheEvent.Type.CHILD_REMOVED)){
                        System.out.println("删除子节点:" + event.getData().getPath());
                        String host = event.getData().getPath().replace("/hosts/","");
                        String  ip = host.split(":")[0];
                        int port = Integer.parseInt(host.split(":")[1]);
                        Channel session = ClientSessions.getInstance().getSessionByIp(ip+":"+port);
                        if(session !=null && session.isActive()) {
                            logger.info("感知到服务端下线，关闭连接");
                            session.close();
                        }
                        //移除连接
                        ClientSessions.getInstance().removeByIp(ip+":"+port);

                    }else if(event.getType().equals(PathChildrenCacheEvent.Type.CHILD_UPDATED)){
                        System.out.println("修改子节点路径:" + event.getData().getPath());
                        System.out.println("修改子节点数据:" + new String(event.getData().getData()));
                    }

                }
            });

            pathChildrenCache.start(PathChildrenCache.StartMode.POST_INITIALIZED_EVENT);

           while(!isConnected){
               Thread.sleep(100);
           }
        } catch (Exception e) {
            System.out.println(e.getLocalizedMessage());
            e.printStackTrace();
        }
    }

    public  <T> T getProxy(Class  serviceInterface) {

        return  (T) Proxy.newProxyInstance(serviceInterface.getClassLoader(), new Class[]{serviceInterface}, new InvocationHandler() {
            @Override
            public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {


                long streamId = IdUtil.nextId();
                //基于响应时间负载均衡选择一台服务器
                Channel clientSession = ClientSessions.getInstance().getSessionByLb(zkclient);


                InetSocketAddress insocket = (InetSocketAddress) clientSession.remoteAddress();
                String host = insocket.getAddress().getHostAddress();
                int port = insocket.getPort();
                logger.info("调用ip:"+host+",port:"+port);

                RpcOperation rpcOperation = new RpcOperation();
                rpcOperation.setClassName(serviceInterface.getName());
                rpcOperation.setMethodName(method.getName());
                rpcOperation.setParameters(args);
                rpcOperation.setParameterTypes(method.getParameterTypes());
                rpcOperation.setHost(host);
                rpcOperation.setPort(port);

                RequestMessage requestMessage = new RequestMessage(
                        streamId, rpcOperation);


                OperationResultFuture operationResultFuture = new OperationResultFuture();

                requestPendingCenter.add(streamId, operationResultFuture);





                long startTime = System.currentTimeMillis();
                clientChannel.channel().writeAndFlush(requestMessage);

                OperationResult operationResult = operationResultFuture.get();

                RpcOperationResult rpcOperationResult = (RpcOperationResult)operationResult;

                //获得本次响应时间
                long responsTime = System.currentTimeMillis() - startTime;

                updateResponseTime(responsTime,rpcOperation);

                if(rpcOperationResult.isSuccess()) {
                    return rpcOperationResult.getResult();
                }else{
                    return rpcOperationResult.getErrors();
                }

            }

            //更新响应时间
            private void updateResponseTime(long responsTime,RpcOperation rpcOperation) {

               String host = rpcOperation.getHost();
               int port = rpcOperation.getPort();

                try {
                    zkclient.setData().forPath("/hosts/"+host+":"+port,(System.currentTimeMillis()+"-"+responsTime).getBytes());
                    logger.error("设置时间成功：{}","/hosts/"+host+":"+port);
                } catch (Exception e) {
                    e.printStackTrace();
                    logger.error("设置时间失败：{}","/hosts/"+host+":"+port);
                }
            }
        });
    }





    public void shutdown(){
        group.shutdownGracefully();
        zkclient.close();
    }




}
